package com.atguigu.edu.app.dim;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.app.BaseApp;
import com.atguigu.edu.bean.TableProcess;
import com.atguigu.edu.common.Constant;
import com.atguigu.edu.util.FlinkSinkUtil;
import com.atguigu.edu.util.JdbcUtil;
import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.Arrays;
import java.util.List;
import java.util.Set;

public class DimApp extends BaseApp {
    public static void main(String[] args) {
        new DimApp().init(2001,1,"DimApp", Constant.TOPIC_ODS_DB);
    }
    @Override
    public void handle(StreamExecutionEnvironment ev, DataStreamSource<String> stream) {
        //对数据进行etl处理
        SingleOutputStreamOperator<JSONObject> etl = etl(stream);
        //使用cdc读取mysql配置文件
        SingleOutputStreamOperator<TableProcess> tpStream = readTableProcess(ev);
        //根据配置信息在hbase中创建维度表
        tpStream = createDimTable(tpStream);
        //合并流，根据广播流的配置信息
        SingleOutputStreamOperator<Tuple2<JSONObject, TableProcess>> dimAndTpStream = connect(etl, tpStream);
        //删除不需要的列
        dimAndTpStream = delNotNeedColumns(dimAndTpStream);
        //将不同的数据写到hbase中
        writeToPhoenix(dimAndTpStream);
   }

    private void writeToPhoenix(SingleOutputStreamOperator<Tuple2<JSONObject, TableProcess>> dimAndTpStream) {


        dimAndTpStream.addSink(FlinkSinkUtil.getPhoenixSink());
    }


    private SingleOutputStreamOperator<Tuple2<JSONObject, TableProcess>> delNotNeedColumns(SingleOutputStreamOperator<Tuple2<JSONObject, TableProcess>> dimAndTpStream) {
        return dimAndTpStream
                .map(new MapFunction<Tuple2<JSONObject, TableProcess>, Tuple2<JSONObject, TableProcess>>() {
                    @Override
                    public Tuple2<JSONObject, TableProcess> map(Tuple2<JSONObject, TableProcess> t) throws Exception {
                        JSONObject data = t.f0;
                        List<String> columns = Arrays.asList(t.f1.getSinkColumns().split(","));

                        Set<String> keys = data.keySet();
                        keys.removeIf(key -> !columns.contains(key) && !"op_type".equals(key));
                        return t;
                    }
                });
    }

    private SingleOutputStreamOperator<Tuple2<JSONObject, TableProcess>> connect(SingleOutputStreamOperator<JSONObject> etl, SingleOutputStreamOperator<TableProcess> tpStream) {

        MapStateDescriptor<String, TableProcess> tpStateDesc = new MapStateDescriptor<>("tpStateDesc", String.class, TableProcess.class);
        BroadcastStream<TableProcess> tpBcStream = tpStream.broadcast(tpStateDesc);

        return etl
                .connect(tpBcStream)
                .process(new BroadcastProcessFunction<JSONObject, TableProcess, Tuple2<JSONObject, TableProcess>>() {
                    // 4. 处理数据流中的数据的时候, 从广播状态获取配置信息
                    @Override
                    public void processElement(JSONObject obj,
                                               ReadOnlyContext ctx,
                                               Collector<Tuple2<JSONObject, TableProcess>> out) throws Exception {
                        ReadOnlyBroadcastState<String, TableProcess> state = ctx.getBroadcastState(tpStateDesc);

                        String key = obj.getString("table") + ":ALL";
                        TableProcess tp = state.get(key);

                        // obj可能是维度数据,也可能是事实表数据, 所以 tp 有可能是空
                        if (tp != null) {
                            JSONObject data = obj.getJSONObject("data");

                            data.put("op_type", obj.getString("type"));
                            out.collect(Tuple2.of(data, tp));

                        }

                    }

                    // 3. 把配置信息放入到广播状态
                    @Override
                    public void processBroadcastElement(TableProcess tp,
                                                        Context ctx,
                                                        Collector<Tuple2<JSONObject, TableProcess>> out) throws Exception {
                        // 获取广播状态
                        BroadcastState<String, TableProcess> state = ctx.getBroadcastState(tpStateDesc);
                        // 放入
                        String key = tp.getSourceTable() + ":" + tp.getSourceType();

                        // 如果 op=d, 表示删除对应的配置信息, 这个时候,也应该删除广播状态中的数据
                        if ("d".equals(tp.getOp())) {
                            state.remove(key);
                        } else {
                            state.put(key, tp);
                        }
                    }
                });
    }

    private SingleOutputStreamOperator<TableProcess> createDimTable(SingleOutputStreamOperator<TableProcess> tpStream) {
        // 只对维度表做操作
        return tpStream
                .filter(tp -> "dim".equals(tp.getSinkType()))
                .process(new ProcessFunction<TableProcess, TableProcess>() {

                    private Connection conn;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        // 1. 建立到 phoenix 的链接
                        conn = JdbcUtil.getPhoenixConnection();
                    }

                    @Override
                    public void close() throws Exception {
                        // 6. 关闭到 phoenix 的链接
                        JdbcUtil.closeConnection(conn);
                    }

                    @Override
                    public void processElement(TableProcess tp,
                                               Context ctx,
                                               Collector<TableProcess> out) throws Exception {
                        String op = tp.getOp();

                        if ("c".equals(op) || "r".equals(op)) {
                            // 建表
                            createTable(tp);
                        } else if ("d".equals(op)) {
                            // 配置信息删除了, 对应的维度表应该删除
                            delTable(tp);
                        } else {
                            // 配置信息发生了变化 u
                            // 1. 先删表
                            delTable(tp);
                            // 2. 再建表
                            createTable(tp);
                        }

                        out.collect(tp);
                    }

                    // 删表
                    private void delTable(TableProcess tp) throws SQLException {
                        // 删表操作
                        // 2. 拼接一个删表的 sql 语句
                        // drop table a
                        String sql = "drop table " + tp.getSinkTable();
                        // 3. 获取一个预处理语句
                        PreparedStatement ps = conn.prepareStatement(sql.toString());
                        ps.execute();
                        ps.close();
                    }

                    // 建表
                    private void createTable(TableProcess tp) throws SQLException {
                        // 建表操作
                        // 2. 拼接一个建表的 sql 语句
                        // create table if not exists user(id varchar, name varchar, constraint pk primary key(id))
                        StringBuffer sql = new StringBuffer();
                        sql
                                .append("create table if not exists ")
                                .append(tp.getSinkTable())
                                .append("(")
                                // 拼接字段
                                .append(tp.getSinkColumns().replaceAll("[^,]+", "$0 varchar"))
                                .append(",constraint pk primary key(")
                                .append(tp.getSinkPk() == null ? "id" : tp.getSinkPk())
                                .append("))")
                                .append(tp.getSinkExtend()==null? "" : tp.getSinkExtend());
                        System.out.println("phoenix 建表语句: " + sql);
                        // 3. 获取一个预处理语句
                        PreparedStatement ps = conn.prepareStatement(sql.toString());
                        // 4. 执行建表语句
                        ps.execute();
                        // 5. 关闭预处理语句
                        ps.close();
                    }
                });
    }

    private SingleOutputStreamOperator<TableProcess> readTableProcess(StreamExecutionEnvironment ev) {
        MySqlSource<String> mySqlSource = MySqlSource.<String>builder()
                .hostname("hadoop162")
                .port(3306)
                .databaseList("edu_config")
                .tableList("edu_config.table_process")
                .username("root")
                .password("aaaaaa")
                .deserializer(new JsonDebeziumDeserializationSchema())
                .build();

        return ev
                .fromSource(mySqlSource, WatermarkStrategy.noWatermarks(), "mysql-cdc")
                .map(new MapFunction<String, TableProcess>() {
                    @Override
                    public TableProcess map(String value) throws Exception {
                        JSONObject obj = JSON.parseObject(value);
                        String op = obj.getString("op");

                        TableProcess tp;
                        if ("d".equals(op)) {
                            // 如果是删除操作, 则获取 before
                            tp = obj.getObject("before", TableProcess.class);
                        } else {
                            // u c r 获取 after
                            tp = obj.getObject("after", TableProcess.class);

                        }
                        tp.setOp(op); // 把 op 的值, 设置到对象中
                        return tp;
                    }
                });
    }

    private SingleOutputStreamOperator<JSONObject> etl(DataStreamSource<String> stream) {
        return stream.filter(new FilterFunction<String>() {
            @Override
            public boolean filter(String value) throws Exception {
                try {
                    JSONObject obj = JSON.parseObject(value);
                    String type = obj.getString("type");
                    String data = obj.getString("data");
                    // 过滤出edu的库，非空的表，只需要插入或者更新的数据，非空的数据
                    return "edu".equals(obj.getString("database"))
                            && null != obj.getString("table")
                            && ("insert".equals(type) || "update".equals(type) || "bootstrap-insert".equals(type))
                            && null != data
                            && data.length() > 2;
                } catch (Exception e) {
                    System.out.println("数据不是合法的 json 数据: " + value);
                    // 解析 json 格式数据出错,表示不是 json 格式
                    return false;
                }
            }
        }).map(json -> JSON.parseObject(json.replace("bootstrap-", "")));
    }
}
